package org.lzfto.flink.demo.common.table;

import org.apache.commons.lang3.ObjectUtils;
import org.lzfto.flink.demo.common.Configuration;

public abstract class PhysicalBasicTable {
	private static String TABLE_DIRECTION_SOURCE = "source";
	private static String TABLE_DIRECTION_TARGET = "target";

	private String env;
	private String direction;
	private String connector;
	private String url;
	private String tableName;
	private String userName;
	private String password;
	private String comments;


	public String getEnv() {
		return env;
	}

	public void setEnv(String env) {
		this.env = env;
	}

	public String getDirection() {
		return direction;
	}

	public void setDirection(String direction) {
		this.direction = direction;
	}

	public String getConnector() {
		return connector;
	}

	public void setConnector(String connector) {
		this.connector = connector;
	}

	public String getUrl() {
		return url;
	}

	public void setUrl(String url) {
		this.url = url;
	}

	public String getTableName() {
		return tableName;
	}

	public void setTableName(String tableName) {
		this.tableName = tableName;
	}

	public String getUserName() {
		return userName;
	}

	public void setUserName(String userName) {
		this.userName = userName;
	}

	public String getPassword() {
		return password;
	}

	public void setPassword(String password) {
		this.password = password;
	}

	public String getComments() {
		return comments;
	}

	public void setComments(String comments) {
		this.comments = comments;
	}
 

	public static PhysicalBasicTable getSourceTable(String tableName) {
		return getTable(TABLE_DIRECTION_SOURCE,tableName);
	}

	public static PhysicalBasicTable getTargetTable(String tableName) {
		return getTable(TABLE_DIRECTION_TARGET,tableName);
	}
	
	private static PhysicalBasicTable getTable(String direction,String tableName) {
		if (ObjectUtils.isEmpty(Configuration.getConf().getTables())) {
			return null;
		}
		
		String env=Configuration.getConf().getEnv();
		for(PhysicalBasicTable table:Configuration.getConf().getTables()) {
			if (table.getEnv().equals(Configuration.getConf().getEnv()) && table.getDirection().equals(direction) &&
					table.getTableName().equals(tableName)) {
				PhysicalBasicTable newTable=null;
				if (table instanceof PhysicalStandardTable ) {
					PhysicalStandardTable tmpTable=new PhysicalStandardTable();
					newTable=tmpTable;
				}
				else if (table instanceof PhysicalCustomJdbcTable custJdbcTable) {
					PhysicalCustomJdbcTable tmpTable=new PhysicalCustomJdbcTable();
					tmpTable.setDriverClass(custJdbcTable.getDriverClass());
					tmpTable.setIdColCode(custJdbcTable.getIdColCode());
					newTable=tmpTable;
					
				}
				newTable.setEnv(table.getEnv());
				newTable.setDirection(table.getDirection());
				newTable.setUrl(table.getUrl());
				newTable.setTableName(tableName);
				newTable.setUserName(table.getUserName());
				newTable.setPassword(table.getPassword());
				newTable.setComments(table.getComments());
				System.out.println("找到表-env="+env+",direction="+direction+",name="+tableName);
				return newTable;
			}
		}
		throw new RuntimeException("没有找到表:"+tableName+"(env="+env+",direction="+direction+")");
	}



	public PhysicalBasicTable withConnector(String connector) {
		this.setConnector(connector);
		return this;
	}

	public abstract String build() throws Exception;

}
